package com.yyy.tcc.service.impl;

import com.mysql.cj.jdbc.MysqlDataSource;
import com.yyy.common.exception.BusinessException;
import com.yyy.tcc.entity.Account;
import com.yyy.tcc.entity.TransferRecord;
import com.yyy.tcc.service.AccountService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.*;
import java.time.LocalDateTime;
import java.time.ZoneId;

public class AccountServiceImpl implements AccountService {
    private static final Logger log = LoggerFactory.getLogger(AccountServiceImpl.class);
    private final DataSource dataSource;

    public AccountServiceImpl() {
        // 初始化数据源
        MysqlDataSource ds = new MysqlDataSource();
        ds.setURL("jdbc:mysql://localhost:3306/distributed_transaction?useSSL=false&serverTimezone=UTC&serverTimezone=Asia/Shanghai");
        ds.setUser("root");
        ds.setPassword("123456");
        this.dataSource = ds;
    }

    @Override
    public boolean tryStageWithTimeout(String transactionId, String fromUserId, String toUserId,
            BigDecimal amount, int timeoutSeconds) {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);

            // 执行普通的tryStage逻辑
            boolean result = executeTryStage(conn, transactionId, fromUserId, toUserId, amount);

            if (result) {
                // 记录超时时间
                String sql = "UPDATE transfer_record SET timeout_time = ? WHERE transaction_id = ?";
                try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                    Timestamp timeoutTime = Timestamp.valueOf(LocalDateTime.now().plusSeconds(timeoutSeconds));
                    pstmt.setTimestamp(1, timeoutTime);
                    pstmt.setString(2, transactionId);
                    pstmt.executeUpdate();
                }
            }

            conn.commit();
            return result;
        } catch (SQLException e) {
            rollback(conn);
            log.error("Try阶段执行失败", e);
            throw new BusinessException("TCC-003", "Try阶段执行失败: " + e.getMessage());
        } finally {
            closeConnection(conn);
        }
    }

    @Override
    public Account getAccount(String userId) {
        String sql = "SELECT * FROM account WHERE user_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setString(1, userId);
            ResultSet rs = pstmt.executeQuery();

            if (rs.next()) {
                Account account = new Account();
                account.setId(rs.getLong("id"));
                account.setUserId(rs.getString("user_id"));
                account.setBalance(rs.getBigDecimal("balance"));
                account.setFrozenAmount(rs.getBigDecimal("freeze_amount"));
                account.setCreateTime(rs.getTimestamp("create_time").toLocalDateTime());
                account.setUpdateTime(rs.getTimestamp("update_time").toLocalDateTime());
                return account;
            }

            throw new BusinessException("TCC-003", "账户不存在: " + userId);

        } catch (SQLException e) {
            log.error("查询账户信息失败: userId={}", userId, e);
            throw new BusinessException("TCC-999", "查询账户信息失败: " + e.getMessage());
        }
    }

    @Override
    public BigDecimal getAccountBalance(String userId) {
        String sql = "SELECT balance FROM account WHERE user_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setString(1, userId);
            ResultSet rs = pstmt.executeQuery();

            if (rs.next()) {
                return rs.getBigDecimal("balance");
            }

            throw new BusinessException("TCC-003", "账户不存在: " + userId);

        } catch (SQLException e) {
            log.error("查询账户余额失败: userId={}", userId, e);
            throw new BusinessException("TCC-999", "查询账户余额失败: " + e.getMessage());
        }
    }

    @Override
    public String getTransactionStatus(String transactionId) {
        String sql = "SELECT status FROM transfer_record WHERE transaction_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setString(1, transactionId);
            ResultSet rs = pstmt.executeQuery();

            if (rs.next()) {
                return rs.getString("status");
            }

            throw new BusinessException("TCC-004", "转账记录不存在: " + transactionId);

        } catch (SQLException e) {
            log.error("查询转账状态失败: transactionId={}", transactionId, e);
            throw new BusinessException("TCC-999", "查询转账状态失败: " + e.getMessage());
        }
    }

    // 实现重试机制的帮助方法
    protected boolean executeWithRetry(String transactionId, RetryableOperation operation, int maxRetries) {
        int retryCount = 0;
        while (retryCount < maxRetries) {
            try {
                return operation.execute();
            } catch (Exception e) {
                retryCount++;
                log.warn("操作失败，第{}次重试", retryCount, e);
                if (retryCount == maxRetries) {
                    log.error("重试次数已达上限", e);
                    throw new BusinessException("TCC-008", "操作失败，重试次数已达上限");
                }
                try {
                    Thread.sleep(1000 * retryCount); // 递增等待时间
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new BusinessException("TCC-009", "重试过程被中断");
                }
            }
        }
        return false;
    }

    // 定义重试操作的函数式接口
    @FunctionalInterface
    protected interface RetryableOperation {
        boolean execute() throws Exception;
    }

    private void rollback(Connection conn) {
        try {
            if (conn != null) {
                conn.rollback();
            }
        } catch (SQLException e) {
            log.error("回滚失败", e);
        }
    }

    private void closeConnection(Connection conn) {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException e) {
            log.error("关闭连接失败", e);
        }
    }

    @Override
    public boolean tryStage(String transactionId, String fromUserId, String toUserId, BigDecimal amount) {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);
            boolean result = executeTryStage(conn, transactionId, fromUserId, toUserId, amount);
            conn.commit();
            return result;
        } catch (SQLException e) {
            rollback(conn);
            log.error("Try阶段执行失败", e);
            throw new BusinessException("TCC-003", "Try阶段执行失败: " + e.getMessage());
        } finally {
            closeConnection(conn);
        }
    }

    // 实现内部的executeTryStage方法
    private boolean executeTryStage(Connection conn, String transactionId, String fromUserId,
            String toUserId, BigDecimal amount) throws SQLException {
        // 幂等性检查
        try (PreparedStatement pstmt = conn.prepareStatement(
                "SELECT COUNT(*) FROM transfer_record WHERE transaction_id = ?")) {
            pstmt.setString(1, transactionId);
            ResultSet rs = pstmt.executeQuery();
            if (rs.next() && rs.getInt(1) > 0) {
                throw new BusinessException("TCC-001", "事务已存在");
            }
        }

        // 检查并锁定账户
        try (PreparedStatement pstmt = conn.prepareStatement(
                "SELECT * FROM account WHERE user_id = ? FOR UPDATE")) {
            pstmt.setString(1, fromUserId);
            ResultSet rs = pstmt.executeQuery();
            if (!rs.next()) {
                throw new BusinessException("TCC-002", "转出账户不存在");
            }
            BigDecimal balance = rs.getBigDecimal("balance");
            if (balance.compareTo(amount) < 0) {
                throw new BusinessException("TCC-002", "余额不足");
            }
        }

        // 冻结金额
        try (PreparedStatement pstmt = conn.prepareStatement(
                "UPDATE account SET balance = balance - ?,freeze_amount = freeze_amount + ? WHERE user_id = ?")) {
            pstmt.setBigDecimal(1, amount);
            pstmt.setBigDecimal(2, amount);
            pstmt.setString(3, fromUserId);
            pstmt.executeUpdate();
        }

        // 创建转账记录
        try (PreparedStatement pstmt = conn.prepareStatement(
                "INSERT INTO transfer_record (transaction_id, from_user_id, to_user_id, amount, status, create_time, update_time) " +
                "VALUES (?, ?, ?, ?, 'TRYING', ?, ?)")) {

            // 获取当前北京时间
            // 修改后：
            Timestamp timestamp = new Timestamp(System.currentTimeMillis());
            pstmt.setString(1, transactionId);
            pstmt.setString(2, fromUserId);
            pstmt.setString(3, toUserId);
            pstmt.setBigDecimal(4, amount);
            pstmt.setTimestamp(5, timestamp);  // create_time
            pstmt.setTimestamp(6, timestamp);  // update_time
            pstmt.executeUpdate();
        }

        return true;
    }

    @Override
    public boolean confirm(String transactionId) {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);

            // 获取并锁定转账记录
            String fromUserId;
            String toUserId;
            BigDecimal amount;
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "SELECT * FROM transfer_record WHERE transaction_id = ? FOR UPDATE")) {
                pstmt.setString(1, transactionId);
                ResultSet rs = pstmt.executeQuery();
                if (!rs.next()) {
                    throw new BusinessException("TCC-004", "转账记录不存在");
                }
                if ("SUCCESS".equals(rs.getString("status"))) {
                    return true; // 幂等性处理
                }
                fromUserId = rs.getString("from_user_id");
                toUserId = rs.getString("to_user_id");
                amount = rs.getBigDecimal("amount");
            }

            // 执行转账
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "UPDATE account SET freeze_amount = freeze_amount - ? " +
                    "WHERE user_id = ?")) {
                pstmt.setBigDecimal(1, amount);
                pstmt.setString(2, fromUserId);
                pstmt.executeUpdate();
            }

            try (PreparedStatement pstmt = conn.prepareStatement(
                    "UPDATE account SET balance = balance + ? WHERE user_id = ?")) {
                pstmt.setBigDecimal(1, amount);
                pstmt.setString(2, toUserId);
                pstmt.executeUpdate();
            }

            // 更新转账状态
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "UPDATE transfer_record SET status = 'SUCCESS' WHERE transaction_id = ?")) {
                pstmt.setString(1, transactionId);
                pstmt.executeUpdate();
            }

            conn.commit();
            return true;
        } catch (SQLException e) {
            rollback(conn);
            log.error("Confirm阶段执行失败", e);
            throw new BusinessException("TCC-005", "Confirm阶段执行失败: " + e.getMessage());
        } finally {
            closeConnection(conn);
        }
    }

    @Override
    public boolean cancel(String transactionId) {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);

            // 获取转账记录
            String fromUserId;
            BigDecimal amount;
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "SELECT * FROM transfer_record WHERE transaction_id = ? FOR UPDATE")) {
                pstmt.setString(1, transactionId);
                ResultSet rs = pstmt.executeQuery();
                if (!rs.next()) {
                    return true; // 记录不存在，视为已取消
                }
                if ("CANCELLED".equals(rs.getString("status"))) {
                    return true; // 幂等性处理
                }
                fromUserId = rs.getString("from_user_id");
                amount = rs.getBigDecimal("amount");
            }

            // 解冻金额
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "UPDATE account SET balance = balance + ?, freeze_amount = freeze_amount - ? WHERE user_id = ?")) {
                pstmt.setBigDecimal(1, amount);
                pstmt.setBigDecimal(2, amount);
                pstmt.setString(3, fromUserId);
                pstmt.executeUpdate();
            }

            // 更新状态
            try (PreparedStatement pstmt = conn.prepareStatement(
                    "UPDATE transfer_record SET status = 'CANCELLED' WHERE transaction_id = ?")) {
                pstmt.setString(1, transactionId);
                pstmt.executeUpdate();
            }

            conn.commit();
            return true;
        } catch (SQLException e) {
            rollback(conn);
            log.error("Cancel阶段执行失败", e);
            throw new BusinessException("TCC-006", "Cancel阶段执行失败: " + e.getMessage());
        } finally {
            closeConnection(conn);
        }
    }

    /**
     * 定时任务：清理超时的Try阶段事务
     */
    public void cleanupTimeoutTransactions() {
        String sql = "SELECT transaction_id FROM transfer_record " +
                    "WHERE status = 'TRYING' AND create_time < ?";

        try (Connection conn = dataSource.getConnection()) {
            conn.setAutoCommit(false);

            try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
                // 获取超过30分钟的事务
                Timestamp timeout = Timestamp.valueOf(LocalDateTime.now().minusMinutes(30));
                pstmt.setTimestamp(1, timeout);

                ResultSet rs = pstmt.executeQuery();
                while (rs.next()) {
                    String transactionId = rs.getString("transaction_id");
                    try {
                        // 尝试取消事务
                        cancel(transactionId);
                        log.info("成功清理超时事务: {}", transactionId);
                    } catch (Exception e) {
                        log.error("清理超时事务失败: {}", transactionId, e);
                        // 记录到补偿队列
                        recordCompensation(transactionId, "CLEANUP_TIMEOUT");
                    }
                }
            }

            conn.commit();
        } catch (SQLException e) {
            log.error("清理超时事务失败", e);
        }
    }

    /**
     * 记录需要补偿的事务
     */
    private void recordCompensation(String transactionId, String reason) {
        String sql = "INSERT INTO compensation_record (transaction_id, reason, status) VALUES (?, ?, 'PENDING')";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setString(1, transactionId);
            pstmt.setString(2, reason);
            pstmt.executeUpdate();

        } catch (SQLException e) {
            log.error("记录补偿失败: {}", transactionId, e);
        }
    }
}
